package pers.mihao.quickstream.common.task.scheduler.isolated;

import com.alibaba.fastjson.JSONObject;
import lombok.extern.slf4j.Slf4j;
import pers.mihao.quickstream.common.task.constans.TaskType;
import pers.mihao.quickstream.common.task.domain.runner.AutomaticTaskRunner;
import pers.mihao.quickstream.common.task.dto.AutomaticTaskDTO;
import pers.mihao.quickstream.common.task.scheduler.TaskInvokeRes;
import pers.mihao.quickstream.common.util.AssertUtil;
import pers.mihao.quickstream.common.util.ThreadPoolNameUtil;

import java.util.Date;
import java.util.concurrent.*;

/**
 * 异步线程有专门的线程池处理
 *
 * @author mihao
 * @since 2024/11/20
 */
@Slf4j
public class AutomaticTaskAsyncRunner implements Runnable {
    AutomaticTaskDTO taskDTO;
    CompletableFuture<TaskInvokeRes> future;
    /**
     * 任务执行交给专门的执行器
     */
    Executor taskInvokerExecutor = new ThreadPoolExecutor(Runtime.getRuntime().availableProcessors(),
            Runtime.getRuntime().availableProcessors() * 3, 10, TimeUnit.MINUTES,
            new ArrayBlockingQueue<>(Runtime.getRuntime().availableProcessors()),
            ThreadPoolNameUtil.getThreadFactory("TTInvoke"),
            new RejectedExecutionHandler() {
                @Override
                public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
                    log.error("线程池处理不完,交给备用线程池");
                    taskInvokerExecutorBack.execute(r);
                }
            });

    Executor taskInvokerExecutorBack = new ThreadPoolExecutor(Runtime.getRuntime().availableProcessors(),
            Runtime.getRuntime().availableProcessors(), 0, TimeUnit.MINUTES,
            new LinkedBlockingQueue<>(),
            ThreadPoolNameUtil.getThreadFactory("TTInvokeBack"));


    public AutomaticTaskAsyncRunner(AutomaticTaskDTO taskDTO,
                                    CompletableFuture<TaskInvokeRes> future) {
        this.taskDTO = taskDTO;
        this.future = future;
    }

    @Override
    public void run() {
        TaskType taskType = taskDTO.getTaskType();
        TaskRunnerAdapter taskRunner = new TaskRunnerAdapter(taskDTO, JSONObject.parseObject(taskDTO.getTaskConfig(),
                AutomaticTaskRunner.class));
        AssertUtil.isTrue(taskRunner.taskType() == taskType);
        try {
            CompletableFuture<TaskInvokeRes> taskFuture = CompletableFuture.supplyAsync(taskRunner::getDoTaskRes, taskInvokerExecutor);
            if (taskDTO.getTimeLimit() != null && taskDTO.getTimeLimit() > 0) {
                taskFuture.orTimeout(taskDTO.getTimeLimit(), TimeUnit.MILLISECONDS);
            }
            taskFuture.whenComplete((invokeSuccess, error) -> {
                if (error != null) {
                    future.completeExceptionally(error);
                } else {
                    future.complete(invokeSuccess);
                }
            });
        } catch (Exception e) {
            log.error("", e);
            future.completeExceptionally(e);
        }
    }

    private static class TaskRunnerAdapter {

        final AutomaticTaskRunner proxy;

        final AutomaticTaskDTO taskDTO;

        public TaskRunnerAdapter(AutomaticTaskDTO taskDTO, AutomaticTaskRunner proxy) {
            this.taskDTO = taskDTO;
            this.proxy = proxy;
        }

        public TaskType taskType() {
            return proxy.taskType();
        }

        public TaskInvokeRes getDoTaskRes() {
            boolean invokeSuccess = false;
            int retryNum = taskDTO.getRetryCount();
            Exception exception = null;
            Date startTaskTime = new Date();
            while (!invokeSuccess && retryNum <= taskDTO.getRetryCount()) {
                try {
                    proxy.doTask();
                } catch (Exception e) {
                    log.error("", e);
                    exception = e;
                } finally {
                    retryNum++;
                }
            }
            return TaskInvokeRes.builder()
                    .taskStartTime(startTaskTime)
                    .error(exception)
                    .invokerRes(invokeSuccess)
                    .taskInvokerNum(retryNum)
                    .taskEndTime(new Date())
                    .build();
        }
    }
}

